# encoding=utf-8
from typing import List, Optional
import json
import asyncio
import logging
import aiohttp
from aiohttp import (
    ClientError,
    ClientResponseError,
)
from aiohttp.client import ClientResponse as Response
from aiohttp.helpers import BasicAuth

from app.libs.json_helper import DateTimeEncoder


from .asynclib import create_task, get_loop
from .common import add_cleanup_task

CONFIG_REQUEST = {
    'is_trust_env': True,
    'is_unsafe_cookiejar': True,  # only meaningful for aiohttp
    'retry_time': 3,
    'retry_interval': 1,
    'request_concurrency': 1000,
    'fetch_timeout': 30,
    'connect_timeout': 15,
    'extra_headers': {},
    'user_agent': r'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36'
}

logger = logging.getLogger(__name__)


async def close_session(session):
    assert isinstance(session, aiohttp.ClientSession)
    if session is not None and not session.closed:
        await session.close()


def new_session(
        user_agent=None,
        is_unsafe_cookiejar=None,
        *,
        connector_kwargs=None,
        loop=None,
        is_add_cleanup=True,
        **session_kwargs,
):
    """Create new aiohttp session."""
    config_request = CONFIG_REQUEST

    if loop is None:
        loop = get_loop()

    if is_unsafe_cookiejar is None:
        is_unsafe_cookiejar = config_request.get('is_unsafe_cookiejar', True)
    cookie_jar = aiohttp.CookieJar(unsafe=is_unsafe_cookiejar, loop=loop)

    if connector_kwargs is None:
        connector_kwargs = {}
    connector_kwargs.setdefault('force_close', True)
    connector_kwargs.setdefault('enable_cleanup_closed', True)
    connector = aiohttp.TCPConnector(loop=loop, **connector_kwargs)

    if session_kwargs is None:
        session_kwargs = {}
    headers = session_kwargs.setdefault('headers', {})
    if user_agent is None:
        user_agent = config_request.get('user_agent')
    assert user_agent
    headers['user-agent'] = user_agent
    session_kwargs['headers'] = headers
    session_kwargs['timeout'] = aiohttp.ClientTimeout(
        total=config_request.get('fetch_timeout', None),
        connect=config_request.get('connect_timeout', None),
    )
    session_kwargs.setdefault('trust_env', bool(config_request.get('is_trust_env')))
    session = aiohttp.ClientSession(
        cookie_jar=cookie_jar,
        connector=connector,
        loop=loop,
        **session_kwargs,
    )

    async def _session_cleanup():
        await close_session(session)

    if is_add_cleanup:
        add_cleanup_task(_session_cleanup)
    return session


class BaseNotifier(object):
    def __init__(
            self,
            url: str,
            method: str,
            auth_type: str = None,
            auth_key: str = None,
            auth_value: str = None,
            add_to: str = None,
            retry_time: int = None,
            retry_interval: float = None,
            connect_timeout: float = None,
    ):
        self._session = None
        self.url = url
        self.method = method
        self.auth_type = auth_type
        self.auth_key = auth_key
        self.auth_value = auth_value
        self.add_to = add_to
        self.retry_time = retry_time or CONFIG_REQUEST.get('retry_time') or 3
        self.retry_interval = retry_interval or CONFIG_REQUEST.get('retry_interval') or 1
        connect_timeout = connect_timeout or CONFIG_REQUEST.get('connect_timeout') or 5
        self.request_default_options = {
            'ssl': False,
            'timeout': connect_timeout,
        }
        self.request_concurrency = int(CONFIG_REQUEST.get('request_concurrency') or 1000)
        assert self.request_concurrency > 0
        self.request_semaphore = asyncio.Semaphore(self.request_concurrency)

    @property
    def session(self):
        if self._session is None:
            headers = {
                'Content-Type': 'application/json; charset=utf8',
            }
            extra_headers = CONFIG_REQUEST['extra_headers']
            if extra_headers:
                assert isinstance(extra_headers, dict)
                headers.update(extra_headers)
            session_map = {"headers": headers}
            if self.auth_type == "basic_auth":
                session_map["auth"] = BasicAuth(login=self.auth_key, password=self.auth_value)
            elif self.auth_type == "api_key" and self.add_to == "header":
                session_map["headers"].update({self.auth_key: self.auth_value})
            self._session = new_session(**session_map)
        return self._session

    async def requests(self, notify_msgs) -> List:
        kwargs = {}
        for key, value in self.request_default_options.items():
            kwargs.setdefault(key, value)
        tasks = []
        for notify_msg in notify_msgs:
            if not notify_msg:
                continue
            tasks.append(create_task(self._request(notify_msg, **kwargs)))
        responses = []
        for task in tasks:
            try:
                responses.append(await task)
            except asyncio.CancelledError:
                raise
            except Exception as err:
                logger.error('url {} error {!r}'.format(self.url, err))
        return responses

    async def _request(self, notify_msg, **kwargs) -> Optional[Response]:
        session = self.session
        method = self.method.upper()
        response = None
        task_id = str(notify_msg["task_id"])
        for retry_time in range(self.retry_time):
            try:
                data = json.dumps({"data": notify_msg}, cls=DateTimeEncoder)
                async with self.request_semaphore:
                    response = await session.request(method, self.url, data=data, **kwargs)
                response.raise_for_status()
            except asyncio.CancelledError:
                raise
            except (asyncio.TimeoutError, TimeoutError) as err:
                logger.warning('{!r}; task_id {} to {} timeout'.format(
                    err, task_id, self.url,
                ))
            except ClientResponseError as err:
                logger.warning('{!r}; task_id {} got error response {}'.format(
                    err, task_id, response,
                ))
            except ClientError as err:
                logger.warning('{!r}; task_id {} failed to connect to {}'.format(
                    err, task_id, self.url
                ))
            else:
                if 200 <= response.status < 300:
                    logger.info('task_id {} to {} success'.format(
                        task_id, self.url,
                    ))
                    return response
                logger.warning('task_id {} to {} got unexpcted response {}'.format(
                    task_id, self.url, response,
                ))
            await asyncio.sleep(self.retry_interval)
        return response

    async def cleanup(self):
        session = self._session
        if session is not None and not session.closed:
            await session.close()

    async def notify(self, notify_msgs: list, is_raise: bool = None):
        is_raise = is_raise or False
        assert isinstance(notify_msgs, list)
        try:
            await self.requests(notify_msgs)
        except asyncio.CancelledError:
            raise
        except Exception as err:
            logger.exception('{!r}; failed to notify'.format(err))
            if is_raise:
                raise
        return None


def get_notifier(*args, **kwargs) -> BaseNotifier:
    default_notifier = BaseNotifier(*args, **kwargs)
    add_cleanup_task(default_notifier.cleanup, 'notifier aiohttp session cleanup')
    return default_notifier
